import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;
import java.io.IOException;



public class DataSort {
    public static class DataSortMap extends Mapper<LongWritable, Text, IntWritable, IntWritable> {
        protected void map(LongWritable key, Text value, Context context)
                throws java.io.IOException, InterruptedException {
            String line = value.toString();
            context.write(new IntWritable(Integer.parseInt(line)), new IntWritable(1));
        }
    }
    public static class DataSortReduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
        private static IntWritable linenum = new IntWritable(1);
        protected void reduce(IntWritable key, Iterable<IntWritable> value, Context context)
                throws java.io.IOException ,InterruptedException {
            for(IntWritable val:value){
                context.write(linenum, key);
                linenum = new IntWritable(linenum.get()+1);
            }
        }
    }
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        try {
            //新建一个Job工作
            Job job = new Job(conf);
            //设置运行类
            job.setJarByClass(DataSort.class);
            //设置要执行的mapper类（自己书写的）
            job.setMapperClass(DataSortMap.class);
            //设置要执行的reduce类（自己书写的）
            job.setReducerClass(DataSortReduce.class);
            //设置输出key的类型
            job.setOutputKeyClass(IntWritable.class);
            //设置输出value的类型
            job.setOutputValueClass(IntWritable.class);

            //设置ruduce任务的个数，默认个数为一个（一般reduce的个数越多效率越高）

            File file = new File("d:/file/output");
            if (file.exists()){
                FileUtil.fullyDelete(file);
            }
            //mapreduce 输入数据的文件/目录
            FileInputFormat.addInputPath(job, new Path("d:/file/input"));
            //mapreduce 执行后输出的数据目录
            FileOutputFormat.setOutputPath(job, new Path("d:/file/output"));

            job.setNumReduceTasks(1);
            //执行完毕退出
            System.exit(job.waitForCompletion(true) ? 0 : 1 );
        } catch (Exception e) {
            e.printStackTrace();
        }
    }



}
